#pragma once

#include "ReactorTaskExecutor.hpp"
#include "ReactorTask.hpp"
#include "../../utility/threadpool/ThreadPool.hpp"
#include "../interface/Event.hpp"
#include <map>
#include <set>

namespace network {

    class IOMultiplexer;

    class Reactor {

    public:

        friend class ReactorTaskExecutor;

        friend class TcpServer;

        friend class ClientConnection;

#if defined(WIN32) || defined(_WIN32)
        friend class Select;
#else
        friend class Epoll;
#endif

        /**
          * @brief  构造一个反应堆
          * @note   默认线程数量为1个
          * @param  pool 线程池智能指针，传入后将被std::move 默认为网络模块设计的线程池
          */
        explicit Reactor(std::shared_ptr<threadpool::ThreadPool> pool = std::make_shared<threadpool::ThreadPool>(
                threadpool::DefaultTaskSchedulerStrategy(), std::make_shared<ReactorTaskExecutor>()));

        ~Reactor() { stop(); }

        /**
          * @brief  为套接字的指定IO事件注册事件处理器到反应堆中
          * @param  fd         要注册的套接字描述符
          * @param  event_type 注册的IO事件类型，允许同时设置多个
          * @param  handler    事件处理器
          * @param  context    用户自定义数据指针，需要用户手动管理内存
          * @retval 是否注册成功
          */
        bool registerHandler(int fd, EventType event_type, const Event::Handler &handler, void *context);

        /**
          * @brief  修改关注的事件
          * @param  fd         要修改的fd
          * @param  event_type 关注的事件
          * @retval 是否成功修改
          */
        bool changeEventType(int fd, EventType event_type);

        /**
        * @brief  删除关心的事件和事件处理器
        * @param  fd 套接字
        */
        void removeHandler(int fd);

        /**
          * @brief  将反应堆内置线程池线程数量添加指定个数
          * @param  num 要添加的线程数量
          */
        void addThread(size_t num) { pool_->add(num); }

        /**
          * @brief  将反应堆内置线程池线程数量删除指定个数
          * @param  num 要删除的线程数量
          */
        void removeThread(size_t num) { pool_->remove(num); }

        /**
          * @brief  向反应堆的等待IO事件哈希集合表中插入一个IO事件
          * @note   线程安全
          * @param  fd         套接字
          * @param  event_type 事件类型，这里只允许设置一种类型，不可以同时设置多个
          */
        void addEvent(int fd, EventType event_type);

        /**
          * @brief  开始事件循环
          * @note   函数为阻塞函数
          */
        void loop();

        /**
          * @brief  关闭反应堆
          */
        void stop() { stop_.store(true); }

        std::mutex ready_event_set_mutex_;

    private:

        // 反应堆关闭标识符
        std::atomic<bool> stop_;

        // 事件集合
        std::map<int, std::shared_ptr<Event>> event_map_;

        // 准备好的事件队列
        std::set<Event *> ready_event_set_;

        // IO多路复用模型
        std::shared_ptr<IOMultiplexer> io_multiplexer_;

        // 反应堆内置线程池
        std::shared_ptr<threadpool::ThreadPool> pool_;

    };

    threadpool::ThreadStatus
    ReactorTaskExecutor::customAfterExecute(const std::shared_ptr<threadpool::Task> &task, threadpool::ThreadPool *pool) {
        if (task->getType() != "ReactorTask") {
            return threadpool::ThreadStatus::ThreadContinue;
        }
        ReactorTask *reactor_task = dynamic_cast<ReactorTask*>(task.get());
        Reactor *reactor = reactor_task->getReactor();
        Event *event = reactor_task->getEvent();
        if (event->empty()) {
            std::lock_guard<std::mutex> lock(reactor->ready_event_set_mutex_);
            reactor->ready_event_set_.erase(reactor_task->getEvent());
        }
        event->setReady();
        return threadpool::ThreadStatus::ThreadContinue;
    }

}

#if defined(WIN32) || defined(_WIN32)
#include "../interface/Select.hpp"
#else
#include "../interface/Epoll.hpp"
#endif

namespace network {

    Reactor::Reactor(std::shared_ptr<threadpool::ThreadPool> pool) : pool_(std::move(pool)), stop_(false) {
#if defined(WIN32) || defined(_WIN32)
        io_multiplexer_ = std::make_shared<Select>();
#else
        io_multiplexer_ = std::make_shared<Epoll>();
#endif
        pool_->add(1);
    }

    bool Reactor::registerHandler(int fd, EventType event_type, const Event::Handler &handler, void *context) {
        event_map_.emplace(fd, std::make_shared<Event>(fd, handler, context));
        return io_multiplexer_->add(fd, event_type);
    }

    bool Reactor::changeEventType(int fd, EventType event_type) {
        return io_multiplexer_->mod(fd, event_type);
    }

    void Reactor::removeHandler(int fd) {
        io_multiplexer_->del(fd);
        event_map_.erase(fd);
    }

    void Reactor::addEvent(int fd, EventType event_type) {
        event_map_.at(fd)->addEventLock(event_type);
        Event *ev = event_map_.at(fd).get();
        ready_event_set_mutex_.lock();
        ready_event_set_.emplace(ev);
        ready_event_set_mutex_.unlock();
        if (!ev->isActive()) {
            ev->setReady();
        }
    }

    void Reactor::loop() {
        stop_.store(false);
        size_t remaining_threads = 0;
        while (!stop_.load()) {
            try {
                io_multiplexer_->dispatch(this);
            }
            catch (const std::exception &e) {
                stop_.store(true);
                break;
            }
            remaining_threads = pool_->threadCount() - pool_->workThreadCount();
            ready_event_set_mutex_.lock();
            auto it = ready_event_set_.begin();
            auto end = ready_event_set_.end();
            while (remaining_threads > 0 && it != end) {
                Event *ev = static_cast<Event*>(*it);
                if (ev->isReady()) {
                    EventType type = ev->popLock();
                    ev->setActive();
                    pool_->submit(std::make_shared<ReactorTask>(
                        std::bind(ev->getHandler(), ev->getFd(), type, ev->getContext()),
                        ev,
                        this));
                    remaining_threads--;
                    ++it;
                }
                else {
                    ++it;
                }
            }
            ready_event_set_mutex_.unlock();
        }
    }

}
